package com.burton.dp;

import java.util.Collections;
import java.util.List;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.CuratorFrameworkFactory.Builder;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;

public class ZookeeperUtil {

	static String nameSpace = "exclusive_lock";
	public static CuratorFramework client;
	static String zkHost = "localhost:2181";
//	static String zkHost = ServiceConfig.get("zookeeper.serverString") + ":2181";

	static {
		RetryPolicy rp = new ExponentialBackoffRetry(1000, 3);
		Builder builder = CuratorFrameworkFactory.builder().connectString("localhost:2181")
				.connectionTimeoutMs(5000)
				.sessionTimeoutMs(5000)
				.retryPolicy(rp);
		builder.namespace(nameSpace);
		client = builder.build();
		client.start();
		client.newNamespaceAwareEnsurePath(nameSpace);
	}

	
	
	public static String create(String zkHost, String path, String content) throws Exception {
		String newNodeName = "";
		client.newNamespaceAwareEnsurePath(path).ensure(client.getZookeeperClient());
		newNodeName = client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, content.getBytes());
		return newNodeName;
	}

	public static void delete(String zkHost, String path) throws Exception {
		client.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
	}

	public static boolean checkExist(String zkHost, String path) throws Exception {
		if (client.checkExists().forPath(path) == null) {
			return false;
		} else {
			return true;
		}
	}

	public static String getChildren(String zkHost, String path) {
		String minNode = "";
		try {
			client.newNamespaceAwareEnsurePath(path).ensure(client.getZookeeperClient());
			List<String> childrens = client.getChildren().forPath("");
			Collections.sort(childrens);
			if (childrens.size() > 1) {
				minNode = childrens.get(1);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return minNode;
	}

	public static String getPreNode(String name, String zkHost, String path) {
		String preNode = "";
		try {
			client.newNamespaceAwareEnsurePath(path).ensure(client.getZookeeperClient());
			List<String> childrens = client.getChildren().forPath("");
			if (childrens.size() > 2) {
				Collections.sort(childrens);
				int cnt = childrens.indexOf(name) - 1;
				if (childrens.get(cnt) != "lock") {
					preNode = childrens.get(cnt);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return preNode;
	}

	public static void main(String[] args) throws Exception {
		
		String path ="/lock";
		if (client.checkExists().forPath(path) == null) {
			client.create().creatingParentsIfNeeded()
					.withMode(CreateMode.PERSISTENT)
					.forPath(path, "int".getBytes());
		}	
		Watcher watcher = new Watcher() {
			public void process(WatchedEvent event) {
				System.out.println("触发回调watcher实例： 路径" + event.getPath() + " 类型："
						+ event.getType());
				if (event.getType().equals(EventType.NodeDeleted)) {

				}
			}
		};
		client.checkExists().usingWatcher(watcher).forPath(path);
		watch();
//		client.delete().forPath(path);
		System.out.println("---------------");
	}

	public static void registerWatcher(Watcher watcher, String watchPath) {
		try {
			client.getData().usingWatcher(watcher).forPath(watchPath);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static void watch() throws Exception {
		NodeCache cache = new NodeCache(client, "/lock", false);
		cache.start();
		System.out.println("监听开始/zk........");
		NodeCacheListener plis = new NodeCacheListener() {
			public void nodeChanged() throws Exception {
				// TODO Auto-generated method stub
				System.out.println("change.................");

			}
		};
		cache.getListenable().addListener(plis);
	}
}
